package edu.mse.camel.client.eventbus.pubsub;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.eclipse.core.runtime.Status;

import edu.mse.camel.client.eventbus.plugin.Activator;
import edu.mse.camel.client.events.IEvent;


/**
 * The EventManager is CamelClient's event bus. Publishers can publish events
 * asynchronously and subscribers can subscribe to multiple events. The order
 * of delivery is not guaranteed.
 * 
 * To subscribe, a subscriber must call EventManager.getInstance().subscribeTo()
 * To publish, a publisher must call 
 * EventManager.getInstance().dispatchEventAsynchronous()
 * 
 * @author darpan
 *
 */
public class EventManager {
	
	static class EventThread extends Thread {
		/**
		 * Queued is a nested top-level (non-member) class. This class
		 * represents the items which are placed on the asynch dispatch queue.
		 * This class is private.
		 */
		private static class QueuedEvent {
			/** object for this event */
			final IEvent event;
			/** next item in event queue */
			QueuedEvent next;
			
			List<ISubscriber> subscribersForEvent;

			/**
			 * Constructor for event queue item
			 *
			 * @param l Listener list for this event
			 * @param d Dispatcher for this event
			 * @param a Action for this event
			 * @param o Object for this event
			 */
			QueuedEvent(IEvent e, List<ISubscriber> subscribers) {
				event = e;
				this.subscribersForEvent = subscribers;
				next = null;
			}
		}
		
		private static int nextThreadNumber;

		private static synchronized String getNextName() {
			return "EventManagerThread-" + nextThreadNumber++; //$NON-NLS-1$
		}
		/** item at the head of the event queue */
		private QueuedEvent head;
		/** item at the tail of the event queue */
		private QueuedEvent tail;

		/** if false the thread must terminate */
		private volatile boolean running;

		/**
		 * Constructor for the event thread.
		 */
		EventThread() {
			this(null, null);
		}

		/**
		 * Constructor for the event thread. 
		 * @param threadName Name of the EventThread 
		 */
		EventThread(String threadName) {
			this(null, threadName);
		}

		/**
		 * Constructor for the event thread. 
		 * @param threadName Name of the EventThread 
		 */
		EventThread(ThreadGroup threadGroup, String threadName) {
			super(threadGroup, threadName == null ? getNextName() : threadName);
			running = true;
			head = null;
			tail = null;
			setDaemon(true); /* Mark thread as daemon thread */
		}

		/**
		 * Stop thread.
		 */
		void close() {
			running = false;
			interrupt();
		}

		/**
		 * This method is called by the thread to remove
		 * items from the queue so that they can be dispatched to their listeners.
		 * If the queue is empty, the thread waits.
		 *
		 * @return The Queued removed from the top of the queue or null
		 * if the thread has been requested to stop.
		 */
		private synchronized QueuedEvent getNextEvent() {
			while (running && (head == null)) {
				try {
					wait();
				} catch (InterruptedException e) {
				}
			}

			if (!running) { /* if we are stopping */
				return null;
			}

			QueuedEvent item = head;
			head = item.next;
			if (head == null) {
				tail = null;
			}

			return item;
		}

		/**
		 * This methods takes the input parameters and creates a Queued
		 * object and queues it.
		 * The thread is notified.
		 *
		 * @param l Listener list for this event
		 * @param d Dispatcher for this event
		 * @param a Action for this event
		 * @param o Object for this event
		 */
		synchronized void queueEvent(List<ISubscriber> subscribersForEvent, IEvent e) {
			if (!isAlive()) { /* If the thread is not alive, throw an exception */
				throw new IllegalStateException();
			}

			QueuedEvent item = new QueuedEvent(e, subscribersForEvent);
			
			if (head == null) /* if the queue was empty */
			{
				head = item;
				tail = item;
			} else /* else add to end of queue */
			{
				tail.next = item;
				tail = item;
			}

			notify();
		}

		/**
		 * This method pulls events from
		 * the queue and dispatches them.
		 */
		public void run() {
			while (true) {
				QueuedEvent item = getNextEvent();
				if (item == null) {
					return;
				}
				if (item.subscribersForEvent != null) {
					for(ISubscriber aSubscriber : item.subscribersForEvent) {
						try {
							aSubscriber.notifyEvent(item.event);
						} catch (Exception e) {
//							if (EventManager.DEBUG) {
								e.printStackTrace();
//							}
						}
					}
				}
			}
		}
	}
	
	static class InitializerThread extends Thread {
		
		private ISubscriber subscriber;
		
		private Iterator<IEvent> events;
		
		public InitializerThread(ISubscriber subscriber, Iterator<IEvent> events) {
			this.subscriber = subscriber;
			this.events = events;
		}
		
		public void run() {
			while (events.hasNext()) {
				subscriber.notifyEvent(events.next());
			}
		}
	}

	static final boolean DEBUG = true;
	
	private static EventManager INSTANCE = new EventManager("Camel event bus dispatcher thread");
	
	/**
	 * @return The singleton instance of the EventManager
	 */
	public static EventManager getInstance() {
		return INSTANCE;
	}
	
	private EventCache cache;
	
	/**
	 * Thread name used for asynchronous event delivery
	 */
	protected final String threadName;
	
	/**
	 * This is a map from event ID to list of subscribers. Each time an event
	 * is published we look into this map to find out who the interested
	 * subscribers are
	 */
	private Map<String, List<ISubscriber>> subscribers;
	
	/**
	 * This is the thread object used for dispatching events.
	 */
	private EventThread thread;
	
	private EventManager(String threadName) {
		subscribers = new HashMap<String, List<ISubscriber>>();
		this.threadName = threadName;
		this.cache = new EventCache();
	}

	/**
	 * For testing purposes only. Do not use this otherwise
	 */
	public void cleanup() {
		subscribers.clear();
	}
	
	/**
	 * This method can be called to release any resources associated with this
	 * EventManager.
	 * <p>
	 * Closing this EventManager while it is asynchronously delivering events 
	 * may cause some events to not be delivered before the async event dispatch 
	 * thread terminates.
	 */
	public synchronized void close() {
		if (thread != null) {
			thread.close();
			thread = null;
		}
	}
	
	/**
	 * This method can be called to publish an event asynchronously. The "publisher"
	 * or caller of this method will return immediately.
	 * 
	 * @param event the event object that needs to be published
	 * @deprecated use {@link EventManager#dispatchEventAsynchronous(IEvent, boolean)}
	 * 
	 */
	public void dispatchEventAsynchronous(IEvent event) {
		EventThread eventThread = this.getEventThread();
		List<ISubscriber> subscribersForEvent = new ArrayList<ISubscriber>(); 
		synchronized (this) {
			//TODO: We need copy-on-write semantics implemented for 
			//the subscribers. Currently we will have trouble if we added subscribers
			//or removed subscribers while dispatching the events.
			List<ISubscriber> subs = subscribers.get(event.getEventID());
			if (subs != null) {
				subscribersForEvent.addAll(subs);
			}
			// Cache the event - for those subscribers who are not awake yet
			cache.cacheEvent(event);
		}
		/* synchronize on the EventThread to ensure no interleaving of posting to the event thread */
		synchronized (eventThread) { 
			eventThread.queueEvent(subscribersForEvent, event);
		}
	}
	
	/**
	 * This method can be called to publish an event asynchronously. The "publisher"
	 * or caller of this method will return immediately.
	 * 
	 * @param event the event object that needs to be published
	 * @param shouldCache true will cache the event
	 */
	public void dispatchEventAsynchronous(IEvent event, boolean shouldCache) {
		EventThread eventThread = this.getEventThread();
		List<ISubscriber> subscribersForEvent = new ArrayList<ISubscriber>(); 
		synchronized (this) {
			//TODO: We need copy-on-write semantics implemented for 
			//the subscribers. Currently we will have trouble if we added subscribers
			//or removed subscribers while dispatching the events.
			List<ISubscriber> subs = subscribers.get(event.getEventID());
			if (subs != null) {
				subscribersForEvent.addAll(subs);
			}
			// Cache the event - for those subscribers who are not awake yet
			if (shouldCache) {
				cache.cacheEvent(event);
			}
		}
		/* synchronize on the EventThread to ensure no interleaving of posting to the event thread */
		synchronized (eventThread) { 
			eventThread.queueEvent(subscribersForEvent, event);
		}
	}
	
	public void dispatchEventSynchronous(IEvent event, boolean getFirstReply) {
		List<ISubscriber> subscribersForEvent = new ArrayList<ISubscriber>(); 
		synchronized (this) {
			List<ISubscriber> subs = subscribers.get(event.getEventID());
			if (subs != null) {
				subscribersForEvent.addAll(subs);
			}
		}
		for (ISubscriber aSubscriber : subscribersForEvent) {
			aSubscriber.notifyEvent(event);
			if (getFirstReply && event.isQueryAnswered()) {
				return;
			}
		}
	}
	
	protected synchronized Iterator<IEvent> getCachedEvents() {
		return cache.getCache();
	}
	
	protected synchronized Iterator<IEvent> getCachedEvents(List<String> eventIDs) {
		return cache.getCache(eventIDs);
	}
	
	/**
	 * Returns the EventThread to use for dispatching events asynchronously for
	 * this EventManager.
	 *
	 * @return EventThread to use for dispatching events asynchronously for
	 * this EventManager.
	 */
	synchronized EventThread getEventThread() {
		if (thread == null) {
			/* if there is no thread, then create a new one */
			thread = new EventThread(threadName);
			thread.start(); /* start the new thread */
		}

		return thread;
	}
	
	/**
	 * For testing purposes only. Do not use this otherwise 
	 */
	public Map<String, List<ISubscriber>> getSubscribers() {
		return subscribers;
	}
	
	/**
	 * The Event manager will send out all events that it has cached so far
	 * since it started. It will also subscribe the subscriber for the events
	 * it is interested in.
	 * 
	 * @param subscriber
	 * @param eventIDs The events you are interested in.
	 * @param getAll get all events or only the ones you are interested in
	 * 
	 * @see {@link EventCache#getCache()} 
	 * @see {@link EventCache#getCache(List)}
	 */
	public synchronized void initializeSubscriber(
			ISubscriber subscriber, List<String> eventIDs, boolean getAll) {
		if (subscriber == null) {
			return;
		}
		for (String anID : eventIDs) {
			subscribeTo(anID, subscriber);
		}
		Iterator<IEvent> events = null;
		if (getAll) {
			events = getCachedEvents();
		} else {
			events = getCachedEvents(eventIDs);
		}
		InitializerThread iThread = new InitializerThread(subscriber, events);
		iThread.start();
	}
	
	/**
	 * This method can be called at anytime by a subscriber to subscribe 
	 * to an event.
	 * It is better to subscribe to all events at once rather than at different places
	 * in your subscriber.
	 * 
	 * @param eventID the event you wish to subscribe to
	 * @param aSubscriber pass a reference to yourself
	 */
	public synchronized void subscribeTo(String eventID, ISubscriber aSubscriber) {
		List<ISubscriber> subscribersForEvent = subscribers.get(eventID);
		if (subscribersForEvent == null) {
			subscribersForEvent = new ArrayList<ISubscriber>();
		}
		else {
			for ( ISubscriber s : subscribersForEvent) {
				if ( s.getClass().getCanonicalName().equals(aSubscriber.getClass().getCanonicalName())) {
					log_impl(Status.WARNING, "Suspicious (duplicated) event subscriber type : " + aSubscriber.getClass().getCanonicalName());
				}
			}
		}
		subscribersForEvent.add(aSubscriber);
		subscribers.put(eventID, subscribersForEvent);
	}
	
	private void log_impl(int level, String msg) {
		Activator.getDefault().getLog().log(new Status(level, Activator.PLUGIN_ID, msg));
	}
	
	/**
	 * This method can be called to remove your subscription of a particular
	 * event.
	 * 
	 * @param eventID the event you wish to unsubscribe to
	 * @param aSubscriber pass yourself
	 * @return true if you were removed
	 */
	public synchronized boolean unSubscribeTo(String eventID, ISubscriber aSubscriber) {
		List<ISubscriber> subscribersForEvent = subscribers.get(eventID);
		if (subscribersForEvent != null) {
			if(! subscribersForEvent.contains(aSubscriber)) {
				log_impl(Status.WARNING, "Unsubscribing error. Subject has not subcribed this event");
			}
			return subscribersForEvent.remove(aSubscriber);
		}
		return false;
	}

	public synchronized void clearCache() {
		cache.clearCache();
	}
	
}
